#pragma once
#include "../common/message.hpp"
#include "../common/net.hpp"
#include "requestor.hpp"

namespace myrpc {
    namespace client {
        class TopicManager {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            using SubCallBack = std::function<void(const std::string &, const std::string &)>;
            TopicManager(const Requestor::ptr &requestor):_requestor(requestor){}

            // 主题创建——向中转服务端发送主题创建请求
            bool topicCreate(const BaseConnection::ptr &conn, const std::string &topic_name){
                return commonRequest(conn, topic_name, TopicOpType::TOPIC_CREATE);
            }
            // 主题删除——向中转服务端发送主题删除请求
            bool topicRemove(const BaseConnection::ptr &conn, const std::string &topic_name){
                return commonRequest(conn, topic_name, TopicOpType::TOPIC_REMOVE);
            }
            // 主题订阅——向中转服务端发送主题订阅请求
            bool topicSubscribe(const BaseConnection::ptr &conn, const std::string &topic_name, const SubCallBack &cb){
                addSubCallBack(topic_name, cb);
                bool ret = commonRequest(conn, topic_name, TopicOpType::TOPIC_SUBSCRIBE);
                if (ret == false){
                    removeSubCallBack(topic_name);
                    return false;
                }
                return true;
            }
            // 主题取消——向中转服务端发送主题取消订阅请求
            bool topicCancel(const BaseConnection::ptr &conn, const std::string &topic_name){
                removeSubCallBack(topic_name);
                return commonRequest(conn, topic_name, TopicOpType::TOPIC_CANCEL);
            }
            // 主题消息发布——向中转服务端发送主题消息发布请求
            bool topicPublish(const BaseConnection::ptr &conn, const std::string &topic_name, const std::string &msg){
                return commonRequest(conn, topic_name, TopicOpType::TOPIC_PUBLISH, msg);
            }

            // 收到订阅的主题发来的消息后的回调处理——注册到Dispatcher消息分发器中
            void onPublish(const BaseConnection::ptr &conn, TopicRequest::ptr &top_req){
                auto optype = top_req->getTopicOpType();
                if (optype != TopicOpType::TOPIC_PUBLISH){
                    client_logger->ERROR("收到了错误类型的主题操作!");
                    return;
                }
                std::string topic_name=top_req->getTopicKey();
                std::string msg = top_req->getTopicMsg();
                SubCallBack callback = getSubCallBack(topic_name);
                if (callback == nullptr){
                    client_logger->ERROR("收到了 %s 主题发来的消息, 但是该消息无主题处理回调", topic_name.c_str());
                    return;
                }
                return callback(topic_name, msg);
            }

        private:
            // 根据不同的主题操作构建主题请求并发送
            bool commonRequest(const BaseConnection::ptr &conn, const std::string &topic_name,
                               TopicOpType optype, const std::string &msg = "")
            {
                auto top_req = MessageFactory::create<TopicRequest>();
                top_req->setMsgId(UuId::uuId());
                top_req->setMsgType(MsgType::REQ_TOPIC);
                top_req->setTopicKey(topic_name);
                top_req->setTopicOptype(optype);
                if (optype == TopicOpType::TOPIC_PUBLISH){
                    top_req->setTopicMsg(msg);
                }
                BaseMessage::ptr msg_req;
                bool ret = _requestor->send(conn, top_req, msg_req);
                if (ret == false){
                    client_logger->ERROR("主题 '%s' 操作请求发送失败!", topic_name.c_str());
                    return false;
                }
                auto top_rsp = std::dynamic_pointer_cast<TopicResponse>(msg_req);
                if (top_rsp.get() == nullptr){
                    client_logger->ERROR("主题 '%s' 操作响应消息转换失败!", topic_name.c_str());
                    return false;
                }
                if (top_rsp->getRCode() != RCode::RCODE_OK){
                    client_logger->ERROR("主题 '%s' 操作失败! 原因: %s", topic_name.c_str(), errReason(top_rsp->getRCode()).c_str());
                    return false;
                }
                return true;
            }
            // 增加新的主题消息的回调处理函数
            void addSubCallBack(const std::string &topic_name, const SubCallBack &cb){
                std::unique_lock<std::mutex> lock(_mutex);
                _topics_cb.insert(std::make_pair(topic_name, cb));
            }
            // 删除目标主题消息的回调处理函数
            void removeSubCallBack(const std::string &topic_name){
                std::unique_lock<std::mutex> lock(_mutex);
                _topics_cb.erase(topic_name);
            }
            // 获取目标主题消息的回调处理函数
            SubCallBack getSubCallBack(const std::string &topic_name){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topics_cb.find(topic_name);
                if (it == _topics_cb.end()){
                    return SubCallBack();
                }
                return it->second;
            }

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, SubCallBack> _topics_cb;  // 建立主题与对应消息的回调函数的映射关系
            Requestor::ptr _requestor;                                // 请求管理对象
        };
    }
}